package com.atguigu.wuliu.app.dwd;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.wuliu.bean.*;
import com.atguigu.wuliu.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

//dwd层订单相关表，根据订单表中的status和 type类型把订单数据进行分流，分为下单，分为下单、支付成功、取消运单行为
//相关维度为小区、地区、用户
//相关度量值为：amount金额、cargo_weight重量
public class DwdOrderReleva {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(5000L);

        //从kafaka读数据
        String topic = "topic_db";
        String group = "dwdOrderGroup";
        DataStreamSource<String> kafakSourceDS = env.fromSource(KafkaUtil.getKafkaSource(topic, group)
                , WatermarkStrategy.noWatermarks(), "kafakSource");

        /*{"database":"wuliu",
        "table":"transport_task",
        "type":"insert",
        "ts":1692240200,
        "xid":347031,
        "commit":true,
        "data":{"id":147791,"shift_id":1089,"line_id":546,"start_org_id":23,"start_org_name":"广东省东莞市转运中心","end_org_id":305,"end_org_name":"广东省东莞市大岭山镇转运站","status":"67002","order_num":0,"driver1_emp_id":1194,"driver1_name":"宇文芸","driver2_emp_id":null,"driver2_name":null,"truck_id":545,"truck_no":"粤BUAYRW","actual_start_time":null,"actual_end_time":null,"actual_distance":null,"create_time":"2023-08-14 22:50:18","update_time":null,"is_deleted":"0"}
        }*/
        //TODO 3.筛选订单和订单明细数据
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafakSourceDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        String tableName = jsonObj.getString("table");
                        jsonObj.remove("database");
                        jsonObj.remove("xid");
                        jsonObj.remove("commit");
                        jsonObj.remove("ts");
                        if ("order_info".equals(tableName) || "order_cargo".equals(tableName)) {
                            out.collect(jsonObj);
                        }
                    }
                }
        );
        //jsonObjDS.prin0t();

       //通过order_id分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(
                new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject jsonObj) throws Exception {
                        String table = jsonObj.getString("table");
                        if ("order_info".equals(table)) {
                            return jsonObj.getJSONObject("data").getString("id");
                        }
                        return jsonObj.getJSONObject("data").getString("order_id");
                    }
                }
        );


           // keyedDS.print();

        //分流，对不同业务进行区分
        // 支付成功明细流标签
        OutputTag<String> paySucTag = new OutputTag<String>("dwd_trade_pay_suc_detail") {
        };
        // 取消运单明细流标签
        OutputTag<String> cancelDetailTag = new OutputTag<String>("dwd_trade_cancel_detail") {
        };
        // 揽收明细流标签
        OutputTag<String> receiveDetailTag = new OutputTag<String>("dwd_trans_receive_detail") {
        };
        // 发单明细流标签
        OutputTag<String> dispatchDetailTag = new OutputTag<String>("dwd_trans_dispatch_detail") {
        };
        // 转运完成明细流标签
        OutputTag<String> boundFinishDetailTag = new OutputTag<String>("dwd_trans_bound_finish_detail") {
        };
        // 派送成功明细流标签
        OutputTag<String> deliverSucDetailTag = new OutputTag<String>("dwd_trans_deliver_detail") {
        };
        // 签收明细流标签
        OutputTag<String> signDetailTag = new OutputTag<String>("dwd_trans_sign_detail") {
        };


        //订单数据先来，先存入状态


        SingleOutputStreamOperator<String> orderDS = keyedDS.process(
                new KeyedProcessFunction<String, JSONObject, String>() {


                    private ValueState<DwdOrderInfoOriginBean> infoBeanState;
                    private ValueState<DwdOrderDetailOriginBean> detailBeanState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<DwdOrderInfoOriginBean> infoBeanStateDescriptor
                                = new ValueStateDescriptor<DwdOrderInfoOriginBean>("infoBeanState", DwdOrderInfoOriginBean.class);
                        infoBeanStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(5)).build());
                        infoBeanState = getRuntimeContext().getState(infoBeanStateDescriptor);

                        ValueStateDescriptor<DwdOrderDetailOriginBean> detailBeanStateDescriptor
                                = new ValueStateDescriptor<DwdOrderDetailOriginBean>("detailBeanState", DwdOrderDetailOriginBean.class);
                        detailBeanState = getRuntimeContext().getState(detailBeanStateDescriptor);
                    }


                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                        //获取表名区分订单表和订单详情表
                        String table = jsonObject.getString("table");
                        //获取数据中的表数据，
                        JSONObject data = jsonObject.getJSONObject("data");
                        //获取度量数据
                        String type = jsonObject.getString("type");
                        //获取状态的值。


                        if ("order_info".equals(table)) {
                            //处理的是订单数据
                            DwdOrderInfoOriginBean infoOriginBean = data.toJavaObject(DwdOrderInfoOriginBean.class);

                            // 脱敏
                            String senderName = infoOriginBean.getSenderName();
                            String receiverName = infoOriginBean.getReceiverName();

                            senderName = senderName.charAt(0) + senderName.substring(1).replaceAll(".", "\\*");
                            receiverName = receiverName.charAt(0) + receiverName.substring(1).replaceAll(".", "\\*");

                            infoOriginBean.setSenderName(senderName);
                            infoOriginBean.setReceiverName(receiverName);

                            DwdOrderDetailOriginBean detailOriginBean = detailBeanState.value();
                            if ("insert".equals(type)) {
                                //下单
                                if (detailOriginBean == null) {
                                    //订单数据 比 订单明细数据先到,将订单数据放到状态中
                                    infoBeanState.update(infoOriginBean);
                                } else {
                                    //说明订单数据来之前，明细数据已经来到了,将订单和明细进行关联
                                    // 属于下单业务过程
                                    DwdTradeOrderDetailBean dwdTradeOrderDetailBean = new DwdTradeOrderDetailBean();
                                    dwdTradeOrderDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                    //将下单业务过程数据 放到主流中
                                    out.collect(JSON.toJSONString(dwdTradeOrderDetailBean));
                                }
                            } else if ("update".equals(type) && detailOriginBean != null) {
                                //下单外的其它操作
                                //获取修改前的数据
                                JSONObject oldData = jsonObject.getJSONObject("old");
                                //获取修改前状态的值
                                String oldStatus = oldData.getString("status");
                                //获取修改后状态的值
                                String status = infoOriginBean.getStatus();
                                if (!oldStatus.equals(status)) {
                                    //说明修改的status字段
                                    String changeLog = oldStatus + " -> " + status;

                                    switch (changeLog) {
                                        case "60010 -> 60020":
                                            // 处理支付成功数据
                                            DwdTradePaySucDetailBean dwdTradePaySucDetailBean = new DwdTradePaySucDetailBean();
                                            dwdTradePaySucDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                            ctx.output(paySucTag, JSON.toJSONString(dwdTradePaySucDetailBean));
                                            break;
                                        case "60020 -> 60030":
                                            // 处理揽收明细数据
                                            DwdTransReceiveDetailBean dwdTransReceiveDetailBean = new DwdTransReceiveDetailBean();
                                            dwdTransReceiveDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                            ctx.output(receiveDetailTag, JSON.toJSONString(dwdTransReceiveDetailBean));
                                            break;
                                        case "60040 -> 60050":
                                            // 处理发单明细数据
                                            DwdTransDispatchDetailBean dispatchDetailBean = new DwdTransDispatchDetailBean();
                                            dispatchDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                            ctx.output(dispatchDetailTag, JSON.toJSONString(dispatchDetailBean));
                                            break;
                                        case "60050 -> 60060":
                                            // 处理转运完成明细数据
                                            DwdTransBoundFinishDetailBean boundFinishDetailBean = new DwdTransBoundFinishDetailBean();
                                            boundFinishDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                            ctx.output(boundFinishDetailTag, JSON.toJSONString(boundFinishDetailBean));
                                            break;
                                        case "60060 -> 60070":
                                            // 处理派送成功数据
                                            DwdTransDeliverSucDetailBean dwdTransDeliverSucDetailBean = new DwdTransDeliverSucDetailBean();
                                            dwdTransDeliverSucDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                            ctx.output(deliverSucDetailTag, JSON.toJSONString(dwdTransDeliverSucDetailBean));
                                            break;
                                        case "60070 -> 60080":
                                            // 处理签收明细数据
                                            DwdTransSignDetailBean dwdTransSignDetailBean = new DwdTransSignDetailBean();
                                            dwdTransSignDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                            ctx.output(signDetailTag, JSON.toJSONString(dwdTransSignDetailBean));
                                            // 签收后订单数据不会再发生变化，状态可以清除
                                            detailBeanState.clear();
                                            break;
                                        default:
                                            if (status.equals("60999")) {
                                                DwdTradeCancelDetailBean dwdTradeCancelDetailBean = new DwdTradeCancelDetailBean();
                                                dwdTradeCancelDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                                ctx.output(cancelDetailTag, JSON.toJSONString(dwdTradeCancelDetailBean));
                                                // 取消后订单数据不会再发生变化，状态可以清除
                                                detailBeanState.clear();
                                            }
                                            break;
                                    }
                                }
                            }
                        } else {//明细表先来
                            //处理的是订单明细数据
                            DwdOrderDetailOriginBean detailOriginBean = data.toJavaObject(DwdOrderDetailOriginBean.class);
                            if ("insert".equals(type)) {
                                //将明细数据放到状态中
                                detailBeanState.update(detailOriginBean);
                                //获取状态中存放的订单数据  注意：只有下单操作，并且订单数据先到，明细数据后到的情况，才会从状态中拿到订单数据
                                DwdOrderInfoOriginBean infoOriginBean = infoBeanState.value();
                                if (infoOriginBean != null) {
                                    //属于下单业务过程
                                    DwdTradeOrderDetailBean dwdTradeOrderDetailBean = new DwdTradeOrderDetailBean();
                                    dwdTradeOrderDetailBean.mergeBean(detailOriginBean, infoOriginBean);
                                    //将下单业务过程数据 放到主流中
                                    out.collect(JSON.toJSONString(dwdTradeOrderDetailBean));
                                }
                            }
                        }

                    }
                }
        );
        SideOutputDataStream<String> paySucDS = orderDS.getSideOutput(paySucTag);
        SideOutputDataStream<String> cancelDetailDS = orderDS.getSideOutput(cancelDetailTag);
        SideOutputDataStream<String> receiveDetailDS = orderDS.getSideOutput(receiveDetailTag);
        SideOutputDataStream<String> dispathDetailDS = orderDS.getSideOutput(dispatchDetailTag);
        SideOutputDataStream<String> boundFinishDetailDS = orderDS.getSideOutput(boundFinishDetailTag);
        SideOutputDataStream<String> deliverSucDetailDS = orderDS.getSideOutput(deliverSucDetailTag);
        SideOutputDataStream<String> signDetailDS = orderDS.getSideOutput(signDetailTag);

       /* orderDS.print("用户下单：");
        paySucDS.print("支付成功：");
        cancelDetailDS.print("取消运单：");
        receiveDetailDS.print("揽收明细：");
        dispathDetailDS.print("发单明细：");
        boundFinishDetailDS.print("转运完成：");
        deliverSucDetailDS.print("派送完成：");
        signDetailDS.print("签收：");*/
        //TODO 9.将不同流的数据写到kafka的不同主题中
        //9.1 定义主题名称
        // 9.1.1 交易域下单明细主题
        String detailTopic = "tms_dwd_trade_order_detail";
        // 9.1.2 交易域支付成功明细主题
        String paySucDetailTopic = "tms_dwd_trade_pay_suc_detail";
        // 9.1.3 交易域取消运单明细主题
        String cancelDetailTopic = "tms_dwd_trade_cancel_detail";
        // 9.1.4 物流域接单（揽收）明细主题
        String receiveDetailTopic = "tms_dwd_trans_receive_detail";
        // 9.1.5 物流域发单明细主题
        String dispatchDetailTopic = "tms_dwd_trans_dispatch_detail";
        // 9.1.6 物流域转运完成明细主题
        String boundFinishDetailTopic = "tms_dwd_trans_bound_finish_detail";
        // 9.1.7 物流域派送成功明细主题
        String deliverSucDetailTopic = "tms_dwd_trans_deliver_detail";
        // 9.1.8 物流域签收明细主题
        String signDetailTopic = "tms_dwd_trans_sign_detail";

        // 9.2 发送数据到 Kafka
        // 9.2.1 运单明细数据
        KafkaSink<String> kafkaProducer = KafkaUtil.getKafkaSink(detailTopic);
        orderDS.print("下单");
        orderDS
                .sinkTo(kafkaProducer)
                .uid("order_detail_sink");

        // 9.2.2 支付成功明细数据
        KafkaSink<String> paySucKafkaProducer = KafkaUtil.getKafkaSink(paySucDetailTopic);
        paySucDS.print("支付成功");
        paySucDS
                .sinkTo(paySucKafkaProducer)
                .uid("pay_suc_detail_sink");

        // 9.2.3 取消运单明细数据
        KafkaSink<String> cancelKafkaProducer = KafkaUtil.getKafkaSink(cancelDetailTopic);
        cancelDetailDS.print("取消运单");
        cancelDetailDS
                .sinkTo(cancelKafkaProducer)
                .uid("cancel_detail_sink");

        // 9.2.4 揽收明细数据
        KafkaSink<String> receiveKafkaProducer = KafkaUtil.getKafkaSink(receiveDetailTopic);
        receiveDetailDS.print("揽收");
        receiveDetailDS
                .sinkTo(receiveKafkaProducer)
                .uid("reveive_detail_sink");

        // 9.2.5 发单明细数据
        KafkaSink<String> dispatchKafkaProducer = KafkaUtil.getKafkaSink(dispatchDetailTopic);
        dispathDetailDS.print("发单");
        dispathDetailDS
                .sinkTo(dispatchKafkaProducer)
                .uid("dispatch_detail_sink");

        // 9.2.6 转运完成明细主题
        KafkaSink<String> boundFinishKafkaProducer = KafkaUtil.getKafkaSink(boundFinishDetailTopic);
        boundFinishDetailDS.print("转运完成");
        boundFinishDetailDS
                .sinkTo(boundFinishKafkaProducer)
                .uid("bound_finish_detail_sink");

        // 9.2.7 派送成功明细数据
        KafkaSink<String> deliverSucKafkaProducer = KafkaUtil.getKafkaSink(deliverSucDetailTopic);
        deliverSucDetailDS.print("派送成功");
        deliverSucDetailDS
                .sinkTo(deliverSucKafkaProducer)
                .uid("deliver_suc_detail_sink");

        // 9.2.8 签收明细数据
        KafkaSink<String> signKafkaProducer = KafkaUtil.getKafkaSink(signDetailTopic);
        signDetailDS.print("签收");
        signDetailDS
                .sinkTo(signKafkaProducer)
                .uid("sign_detail_sink");


        env.execute();

    }
}

